package com.usefullc.crawler.task.impl;

import com.usefullc.common.enums.YesNoEnum;
import com.usefullc.common.exception.BizException;
import com.usefullc.crawler.constants.CrawlerCacheConstants;
import com.usefullc.crawler.task.TaskMessage;
import com.usefullc.crawler.task.TaskMessageQueue;
import com.usefullc.crawler.vo.TaskTypeVo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 都只考虑一个线程操作
 *
 * @author: Sampson
 * @date: 2022-03-27 16:39
 */
@Service("defaultTaskMessageQueue")
public class DefaultTaskMessageQueue<T> implements TaskMessageQueue<T> {


    private BlockingQueue<TaskMessage> taskQueue = new LinkedBlockingQueue<>();

    private Map<String, BlockingQueue<TaskMessage<T>>> taskMap = new ConcurrentHashMap<>();

    @Autowired
    private RedisTemplate redisTemplate;



    @Override
    public void addTaskMessage(TaskMessage<T> taskMessage) {
        //taskQueue.offer(taskMessage);
        String queueKey = CrawlerCacheConstants.CRAWLER_TASK_QUEUE +"_"+taskMessage.getTaskId();
        redisTemplate.opsForList().rightPush(queueKey,taskMessage);

//    String key = taskMessage.getReqType();
//    MapUtils.putConcurrentListElement(taskMap,key,taskMessage);

    }

    @Override
    public TaskMessage<T> popTaskMessage(boolean block) {
//    BlockingQueue<TaskMessage> taskMessages = taskMap.get(reqType);
        //暂时废弃
        if(true){
            throw new BizException("no implements");
        }
        if (block) {
            try {
                return taskQueue.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

        } else {
            return taskQueue.poll();

        }
    }

    @Override
    public List<TaskMessage<T>> popTaskMultiMessage(boolean block) {
        //多个节点分布式锁 todo
        List<TaskMessage<T>> resultList = new ArrayList<>();
        long currentTimeMillis = System.currentTimeMillis();
        //先获取任务类型队列
        HashOperations<String, String, TaskTypeVo> hashOps = redisTemplate.opsForHash();

        String taskTypeKey = CrawlerCacheConstants.CRAWLER_TASK_TYPE;
        Map<String, TaskTypeVo> taskTypeMap = hashOps.entries(taskTypeKey);
        for (Map.Entry<String, TaskTypeVo> entry : taskTypeMap.entrySet()) {
            TaskTypeVo activeTypeVo = entry.getValue();
            String activeProcessId = entry.getKey();
            if (activeTypeVo.getStatus() != null && activeTypeVo.getStatus() == YesNoEnum.NO.getCode()) {  //任务锁死，则直接返回
                activeTypeVo.setLastTime(System.currentTimeMillis());  //修改最后访问时间为当前时间，防止任务堆积
                redisTemplate.opsForHash().put(CrawlerCacheConstants.CRAWLER_TASK_TYPE, activeProcessId, activeTypeVo);
                continue;
            }
            Long lastTime = activeTypeVo.getLastTime();
            Long waitTime = activeTypeVo.getWaitTime();
            if(currentTimeMillis - lastTime < waitTime){  //获取时间还没到
                continue;
            }
            String queueKey = CrawlerCacheConstants.CRAWLER_TASK_QUEUE +"_"+activeTypeVo.getTaskId();
            TaskMessage taskMessage = (TaskMessage) redisTemplate.opsForList().leftPop(queueKey);
            if(taskMessage != null){
                resultList.add(taskMessage);
                //修改activeType
                activeTypeVo.setLastTime(currentTimeMillis);
                redisTemplate.opsForHash().put(taskTypeKey,entry.getKey(),activeTypeVo);
            }
        }

        return resultList;
    }

    @Override
    public List<TaskMessage<T>> popTaskMultiMessage(Long taskId) {
        //多个节点分布式锁 todo
        List<TaskMessage<T>> resultList = new ArrayList<>();
        long currentTimeMillis = System.currentTimeMillis();
        //先获取任务类型队列
        HashOperations<String, String, TaskTypeVo> hashOps = redisTemplate.opsForHash();

        String taskTypeKey = CrawlerCacheConstants.CRAWLER_TASK_TYPE;
        Map<String, TaskTypeVo> taskTypeMap = hashOps.entries(taskTypeKey);
        for (Map.Entry<String, TaskTypeVo> entry : taskTypeMap.entrySet()) {
            TaskTypeVo activeTypeVo = entry.getValue();
            if(activeTypeVo.getTaskId().longValue() != taskId){
                continue;
            }
            if (activeTypeVo.getStatus() != null && activeTypeVo.getStatus() == YesNoEnum.NO.getCode()) {  //任务锁死，则直接返回
                continue;
            }
            Long lastTime = activeTypeVo.getLastTime();
            Long waitTime = activeTypeVo.getWaitTime();
            if(currentTimeMillis - lastTime < waitTime){  //获取时间还没到
                continue;
            }
            String queueKey = CrawlerCacheConstants.CRAWLER_TASK_QUEUE +"_"+activeTypeVo.getTaskId();
            TaskMessage taskMessage = (TaskMessage) redisTemplate.opsForList().leftPop(queueKey);
            if(taskMessage != null){
                resultList.add(taskMessage);
                //修改activeType
                activeTypeVo.setLastTime(currentTimeMillis);
                redisTemplate.opsForHash().put(taskTypeKey,entry.getKey(),activeTypeVo);
            }
        }

        return resultList;
    }
//  @Override
//  public TaskMessage popTaskMessage(String reqType) {
//    BlockingQueue<TaskMessage> taskMessages = taskMap.get(reqType);
//    try {
//      return taskMessages.take();
//    } catch (InterruptedException e) {
//      throw new RuntimeException(e);
//    }
//  }

//  @Override
//  public List<TaskMessage> popTaskMessages(String reqType, int length) {
//    BlockingQueue<TaskMessage> taskMessages = taskMap.get(reqType);
//    if(taskMessages.size() < length){
//      length = taskMessages.size();
//    }
//    List<TaskMessage> dataList = new ArrayList<>();
//    Iterator<TaskMessage> iterator = taskMessages.iterator();
//    while(iterator.hasNext() && length > 0){
//      dataList.add(iterator.next());
//      length--;
//    }
//    return dataList;
//  }
//
//  @Override
//  public void commit(TaskMessage taskMessage) {
//    BlockingQueue<TaskMessage> taskMessages = taskMap.get(taskMessage.getReqType());
//    taskMessages.remove(taskMessage);
//  }
//
//  @Override
//  public void commit(String reqType, int length) {
//    BlockingQueue<TaskMessage> taskMessages = taskMap.get(reqType);
//    if (taskMessages.size() < length) {
//      throw new IndexOutOfBoundsException("taskMessages  size="+taskMessages.size()+",commit length="+length);
//    }
//    while(length > 0){
//      taskMessages.remove(0);
//      length--;
//    }
//
//  }

}
